package com.mo.axion.stream.utils;

import com.mo.axion.stream.constants.AxionFlinkConstant;
import com.mo.axion.stream.callback.RedisPipelineCallBack;
import io.lettuce.core.LettuceFutures;
import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisFuture;
import io.lettuce.core.RedisURI;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.async.RedisAsyncCommands;
import io.lettuce.core.codec.ByteArrayCodec;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author Mo
 * @createTime 2024/4/13 17:37
 * @description 基于Lettuce封装Redis操作
 */
public class LettuceRedisUtils {
    private static RedisClient redisClient;
    /**
     * 初始化 redisClient
     */
    static {
        RedisURI redisURI = RedisURI.Builder.redis(AxionFlinkConstant.REDIS_IP)
                .withPort(Integer.parseInt(AxionFlinkConstant.REDIS_PORT))
                .withPassword(AxionFlinkConstant.REDIS_PASSWORD.toCharArray())
                .build();
        redisClient = RedisClient.create(redisURI);
    }

    private LettuceRedisUtils() {

    }

    /**
     * 封装pipeline操作
     */
    public static void pipeline(RedisPipelineCallBack pipelineCallBack) {
        StatefulRedisConnection<byte[], byte[]> connect = redisClient.connect(new ByteArrayCodec());
        RedisAsyncCommands<byte[], byte[]> commands = connect.async();

        List<RedisFuture<?>> futures = pipelineCallBack.invoke(commands);

        commands.flushCommands();
        LettuceFutures.awaitAll(10, TimeUnit.SECONDS,
                futures.toArray(new RedisFuture[futures.size()]));
        connect.close();
    }
}
